[kafka] Define DDL table mapping for Kafka compatibility - #4277

Purpose

Closes #4275. Part of #4185.

Define the shared DDL table mapping contract used by Metadata and Produce. A Kafka topic named database.table resolves directly to that Fluss table, so tables with the same name in different databases remain distinct. The physical schema and kafka.* custom properties define how record fields map to table columns.

This is PR03 in the foundation sequence, based on PR02 (#4260, commit 5742b36627fe156a8cf499f0aa7b4b8c4163e9a7); Metadata follows as PR04 (#4261).

Brief change log

• Map fully qualified topic names in both directions and derive stable topic IDs from Fluss table IDs. Validate both Fluss component names and Kafka's full topic-name length limit.
• Validate non-partitioned Arrow log tables without primary keys and resolve raw/string key and value formats.
• Resolve key/value field projections and optional timestamp/headers columns; reject invalid types, missing fields, overlapping projections, and unsupported mapping options. Build projected field names without repeatedly allocating the full schema's field-name list.
• Cover naming boundaries, schema validation, and native Fluss create-table → metadata-readback behavior, including same-named tables in different databases. Reuse the existing curator-test version in test scope for the cluster fixture.

Relative to PR02: 10 files, +1374 / -0 lines (791 production/build and 583 test lines). The change keeps the mapping contract and its validation together. Record decoding, Arrow encoding, Metadata handlers, Produce, JSON, and authentication follow in later PRs.

Tests

Validated on Java 11 with the source tree committed as 4361fa11b227f6d5573757b21499f692a289cc12:

• 67 Kafka unit tests passed using the same-checkout Maven reactor.
• KafkaRequestITCase and KafkaTableMappingITCase passed: 2 integration tests, including cross-database mapping and native metadata readback.
• Checkstyle, Spotless, RAT, and git diff --check passed.

The full repository test suite and a Flink SQL runtime were not run. The later delivery stage will cover the full DDL → Metadata → Kafka producer → Fluss readback workflow.

API and Format

Uses the existing table schema and custom-property storage. kafka.value.format is required; key mapping is optional. Each mapped raw or string component requires exactly one BYTES or STRING column respectively. Configured key fields require kafka.value.fields-include=EXCEPT_KEY; timestamp and headers use explicitly named physical columns and are excluded from the value projection. Column names are user-defined.

Topic names must use database.table; bare table names do not fall back to a configured database. Reverse mapping returns the same fully qualified name.

No Kafka API is newly advertised by this PR. Table creation continues through existing Fluss DDL/Admin or the Flink catalog; the resolver validates the stored mapping when called by subsequent API layers. Kafka CreateTopics/DeleteTopics and automatic creation are out of scope.

Documentation

The contract is described in this PR, source Javadoc, and executable tests. User-facing delivery examples follow with the later Produce capability.

Stack and review

1. [kafka] Add request dispatch and transport framework #4259 — PR01: request and transport framework.
2. [kafka] Serve ApiVersions from registered capabilities #4260 — PR02: ApiVersions.
3. This PR — PR03: DDL table mapping ([kafka] Define DDL table mapping for Kafka compatibility #4275).
4. [kafka] Implement Metadata API over Fluss table metadata #4261 — PR04: Metadata ([kafka] Implement Metadata API for pre-created Fluss tables #4266).

This PR remains a draft while its prerequisites are unmerged. It targets Apache main, so the Files changed view includes the foundation prerequisites. Review DDL mapping alone using the incremental comparison against PR02.

Generative AI disclosure: Codex assisted with implementation, testing, and commit organization. Human review is required before merge.